//! # 流处理管道实现
//! 
//! 提供流处理管道的具体实现，管理整个数据流处理流程

use async_trait::async_trait;
use std::sync::{Arc, Weak};
use std::collections::HashMap;
use std::time::{Duration, Instant};
use tokio::sync::{RwLock, mpsc, oneshot, Mutex};
use tokio::time::{sleep, timeout};
use tokio_stream::{wrappers::UnboundedReceiverStream};
use tracing::{debug, info, warn, error, instrument};
use uuid::Uuid;
use serde_json::json;

use rustcloud_bus::prelude::Message;
use crate::core::*;
use crate::error::{StreamError, StreamResult};
use crate::config::{StreamConfig, PerformanceConfig, BackpressureStrategy};

/// 默认流处理管道实现
#[derive(Debug)]
pub struct DefaultStreamPipeline {
    /// 管道名称
    name: String,
    /// 管道配置
    config: StreamConfig,
    /// 数据源
    source: Arc<Mutex<dyn StreamSource<Item = Message>>>,
    /// 数据汇
    sink: Arc<Mutex<dyn StreamSink<Message>>>,
    /// 处理器链
    processors: RwLock<Vec<Arc<dyn StreamProcessor<Message, Message>>>>,
    /// 管道状态
    status: RwLock<PipelineStatus>,
    /// 统计信息
    stats: RwLock<PipelineStats>,
}

/// 管道控制命令
#[derive(Debug, Clone)]
enum PipelineCommand {
    Start,
    Stop,
    Pause,
    Resume,
    AddProcessor(Arc<dyn StreamProcessor<Message, Message>>),
    RemoveProcessor(String),
    UpdateConfig(StreamConfig),
}

impl DefaultStreamPipeline {
    /// 创建新的流处理管道
    pub fn new(
        name: impl Into<String>,
        config: StreamConfig,
        source: Arc<Mutex<dyn StreamSource<Item = Message>>>,
        sink: Arc<Mutex<dyn StreamSink<Message>>>,
    ) -> Self {
        let mut stats = PipelineStats::new();
        stats.start_time = Some(chrono::Utc::now());

        Self {
            name: name.into(),
            config,
            source,
            sink,
            processors: RwLock::new(Vec::new()),
            status: RwLock::new(PipelineStatus::Uninitialized),
            stats: RwLock::new(stats),
        }
    }

    /// 处理单个消息
    #[instrument(skip(self, message, context))]
    async fn process_message(&self, message: Message, context: StreamContext) -> StreamResult<Option<Message>> {
        let start_time = Instant::now();
        let mut current_message = message;
        
        debug!("开始处理消息: {}", context.message_id);

        // 依次通过处理器链
        let processors = self.processors.read().await;
        for (index, processor) in processors.iter().enumerate() {
            match processor.process(current_message.clone()).await {
                Ok(processed_message) => {
                    current_message = processed_message;
                    debug!("处理器 {} 处理成功", processor.name());
                }
                Err(e) => {
                    error!("处理器 {} 处理失败: {}", processor.name(), e);
                    
                    // 根据错误处理策略进行处理
                    match self.handle_processing_error(&e, &current_message, index).await {
                        ErrorHandlingResult::Retry => {
                            // 重试逻辑
                            match processor.process(current_message.clone()).await {
                                Ok(processed_message) => {
                                    current_message = processed_message;
                                    warn!("处理器 {} 重试成功", processor.name());
                                }
                                Err(retry_error) => {
                                    error!("处理器 {} 重试失败: {}", processor.name(), retry_error);
                                    return Err(retry_error);
                                }
                            }
                        }
                        ErrorHandlingResult::Skip => {
                            warn!("跳过错误，继续处理");
                            continue;
                        }
                        ErrorHandlingResult::Abort => {
                            error!("中止处理");
                            return Err(e);
                        }
                        ErrorHandlingResult::DeadLetter => {
                            warn!("发送到死信队列");
                            self.send_to_dead_letter(&current_message, &e).await?;
                            return Ok(None);
                        }
                    }
                }
            }
        }

        // 更新统计信息
        let processing_time = start_time.elapsed().as_millis() as f64;
        let mut stats = self.stats.write().await;
        stats.record_success(processing_time);

        debug!("消息处理完成: {}, 耗时: {}ms", context.message_id, processing_time);
        Ok(Some(current_message))
    }

    /// 错误处理
    async fn handle_processing_error(
        &self,
        error: &StreamError,
        _message: &Message,
        _processor_index: usize,
    ) -> ErrorHandlingResult {
        // 简化的错误处理逻辑，实际应该根据配置决定
        match error {
            StreamError::TimeoutError { .. } => ErrorHandlingResult::Retry,
            StreamError::NetworkError { .. } => ErrorHandlingResult::Retry,
            StreamError::ProcessorError { .. } => ErrorHandlingResult::DeadLetter,
            _ => ErrorHandlingResult::Abort,
        }
    }

    /// 发送到死信队列
    async fn send_to_dead_letter(&self, message: &Message, error: &StreamError) -> StreamResult<()> {
        let dlq_config = &self.config.error_handling.dead_letter;
        if !dlq_config.enabled {
            return Ok(());
        }

        // 创建死信消息
        let mut dlq_message = message.clone();
        
        // 添加错误信息到元数据
        dlq_message.metadata_mut().set_header("x-dlq-reason".to_string(), error.to_string());
        dlq_message.metadata_mut().set_header("x-dlq-timestamp".to_string(), chrono::Utc::now().to_rfc3339());
        dlq_message.metadata_mut().set_header("x-dlq-original-topic".to_string(), message.topic().to_string());
        
        // 这里应该发送到实际的死信队列
        warn!("消息发送到死信队列: {} -> {}", message.topic(), dlq_config.topic);
        
        Ok(())
    }

    /// 主工作循环
    #[instrument(skip(self, control_rx, shutdown_rx))]
    async fn worker_loop(
        &self,
        mut control_rx: mpsc::UnboundedReceiver<PipelineCommand>,
        mut shutdown_rx: oneshot::Receiver<()>,
    ) {
        info!("流处理管道 {} 开始工作", self.name);
        
        let mut is_paused = false;
        let _performance_config = &self.config.performance;
        
        // 创建消息处理通道
        let (message_tx, mut message_rx) = mpsc::unbounded_channel::<Message>();
        
        // 启动源读取任务
        let source_handle = self.start_source_reader(message_tx.clone()).await;
        
        loop {
            tokio::select! {
                // 处理控制命令
                command = control_rx.recv() => {
                    match command {
                        Some(PipelineCommand::Pause) => {
                            is_paused = true;
                            info!("管道已暂停");
                            *self.status.write().await = PipelineStatus::Paused;
                        }
                        Some(PipelineCommand::Resume) => {
                            is_paused = false;
                            info!("管道已恢复");
                            *self.status.write().await = PipelineStatus::Running;
                        }
                        Some(PipelineCommand::Stop) => {
                            info!("收到停止命令");
                            break;
                        }
                        Some(PipelineCommand::AddProcessor(processor)) => {
                            self.processors.write().await.push(processor);
                            info!("已添加处理器");
                        }
                        Some(PipelineCommand::RemoveProcessor(name)) => {
                            let mut processors = self.processors.write().await;
                            processors.retain(|p| p.name() != name);
                            info!("已移除处理器: {}", name);
                        }
                        None => {
                            warn!("控制通道已关闭");
                            break;
                        }
                        _ => {}
                    }
                }
                
                // 处理停止信号
                _ = &mut shutdown_rx => {
                    info!("收到停止信号");
                    break;
                }
                
                // 处理消息
                message = message_rx.recv(), if !is_paused => {
                    match message {
                        Some(msg) => {
                            let context = StreamContext::new(&self.name, msg.id().to_string());
                            
                            match self.process_message(msg, context).await {
                                Ok(Some(processed_msg)) => {
                                    // 发送到输出汇
                                    if let Err(e) = self.sink.lock().await.send(processed_msg).await {
                                        error!("发送到输出汇失败: {}", e);
                                        self.stats.write().await.record_failure();
                                    }
                                }
                                Ok(None) => {
                                    // 消息被过滤或发送到死信队列
                                    debug!("消息被过滤");
                                }
                                Err(e) => {
                                    error!("消息处理失败: {}", e);
                                    self.stats.write().await.record_failure();
                                }
                            }
                        }
                        None => {
                            warn!("消息通道已关闭");
                            break;
                        }
                    }
                }
                
                // 定期更新统计信息
                _ = sleep(Duration::from_secs(1)) => {
                    self.stats.write().await.calculate_throughput();
                }
            }
        }
        
        // 清理工作
        if let Some(handle) = source_handle {
            handle.abort();
        }
        
        info!("流处理管道 {} 工作循环结束", self.name);
    }

    /// 启动源读取器
    async fn start_source_reader(
        &self,
        message_tx: mpsc::UnboundedSender<Message>,
    ) -> Option<tokio::task::JoinHandle<()>> {
        let source = Arc::downgrade(&self.source);
        let name = self.name.clone();
        
        Some(tokio::spawn(async move {
            info!("启动源读取器: {}", name);
            
            while let Some(source_arc) = source.upgrade() {
                let mut source_guard = source_arc.lock().await;
                
                match source_guard.next().await {
                    Some(Ok(message)) => {
                        if message_tx.send(message).is_err() {
                            warn!("消息通道已关闭，停止源读取器");
                            break;
                        }
                    }
                    Some(Err(e)) => {
                        error!("从源读取消息失败: {}", e);
                        sleep(Duration::from_millis(1000)).await;
                    }
                    None => {
                        if source_guard.is_exhausted() {
                            info!("数据源已耗尽");
                            break;
                        }
                        sleep(Duration::from_millis(100)).await;
                    }
                }
                
                drop(source_guard); // 释放锁
                tokio::task::yield_now().await; // 让出控制权
            }
            
            info!("源读取器结束: {}", name);
        }))
    }
}

#[async_trait]
impl StreamPipeline for DefaultStreamPipeline {
    #[instrument(skip(self))]
    async fn start(&self) -> StreamResult<()> {
        let mut status = self.status.write().await;
        
        match *status {
            PipelineStatus::Running => {
                return Err(StreamError::internal_error("管道已在运行中"));
            }
            PipelineStatus::Starting => {
                return Err(StreamError::internal_error("管道正在启动"));
            }
            _ => {}
        }

        *status = PipelineStatus::Starting;
        drop(status);

        info!("启动流处理管道: {}", self.name);

        // 验证配置
        self.config.validate()?;

        // 启动源和汇
        self.source.lock().await.start().await
            .map_err(|e| StreamError::initialization_error(format!("启动数据源失败: {}", e)))?;

        // 更新状态
        *self.status.write().await = PipelineStatus::Running;
        self.stats.write().await.start_time = Some(chrono::Utc::now());

        info!("流处理管道 {} 启动成功", self.name);
        Ok(())
    }

    #[instrument(skip(self))]
    async fn stop(&self) -> StreamResult<()> {
        let mut status = self.status.write().await;
        
        if matches!(*status, PipelineStatus::Stopped | PipelineStatus::Stopping) {
            return Ok(());
        }

        *status = PipelineStatus::Stopping;
        drop(status);

        info!("停止流处理管道: {}", self.name);

        // 停止源和汇
        self.source.lock().await.stop().await?;
        self.sink.lock().await.close().await?;

        // 更新状态
        *self.status.write().await = PipelineStatus::Stopped;

        info!("流处理管道 {} 已停止", self.name);
        Ok(())
    }

    #[instrument(skip(self))]
    async fn pause(&self) -> StreamResult<()> {
        let mut status = self.status.write().await;
        if !matches!(*status, PipelineStatus::Running) {
            return Err(StreamError::internal_error("只有运行中的管道可以暂停"));
        }
        *status = PipelineStatus::Paused;
        info!("管道 {} 已暂停", self.name);
        Ok(())
    }

    #[instrument(skip(self))]
    async fn resume(&self) -> StreamResult<()> {
        let mut status = self.status.write().await;
        if !matches!(*status, PipelineStatus::Paused) {
            return Err(StreamError::internal_error("只有暂停的管道可以恢复"));
        }
        *status = PipelineStatus::Running;
        info!("管道 {} 已恢复", self.name);
        Ok(())
    }

    fn name(&self) -> &str {
        &self.name
    }

    async fn status(&self) -> StreamResult<PipelineStatus> {
        Ok(self.status.read().await.clone())
    }

    async fn stats(&self) -> StreamResult<PipelineStats> {
        let mut stats = self.stats.read().await.clone();
        stats.calculate_throughput();
        Ok(stats)
    }

    async fn wait_for_completion(&self) -> StreamResult<()> {
        // 简化实现：等待直到状态不再是 Running
        loop {
            let status = self.status.read().await;
            if !matches!(*status, PipelineStatus::Running) {
                break;
            }
            drop(status);
            sleep(Duration::from_millis(100)).await;
        }
        Ok(())
    }

    fn add_processor(&mut self, processor: Arc<dyn StreamProcessor<Message, Message>>) -> StreamResult<()> {
        // 简化实现：直接添加到列表（在实际实现中需要考虑线程安全）
        info!("添加处理器: {}", processor.name());
        Ok(())
    }

    fn remove_processor(&mut self, name: &str) -> StreamResult<()> {
        // 简化实现：直接返回成功（在实际实现中需要考虑线程安全）
        info!("移除处理器: {}", name);
        Ok(())
    }

    fn processors(&self) -> Vec<String> {
        // 这里需要在实际实现中返回处理器名称列表
        // 由于异步锁的限制，这里返回空列表
        Vec::new()
    }
}

/// 错误处理结果
#[derive(Debug, Clone)]
enum ErrorHandlingResult {
    Retry,
    Skip,
    Abort,
    DeadLetter,
}

/// 流处理管道构建器
pub struct StreamPipelineBuilder {
    name: String,
    config: Option<StreamConfig>,
    source: Option<Arc<Mutex<dyn StreamSource<Item = Message>>>>,
    sink: Option<Arc<Mutex<dyn StreamSink<Message>>>>,
    processors: Vec<Arc<dyn StreamProcessor<Message, Message>>>,
}

impl StreamPipelineBuilder {
    /// 创建新的构建器
    pub fn new(name: impl Into<String>) -> Self {
        Self {
            name: name.into(),
            config: None,
            source: None,
            sink: None,
            processors: Vec::new(),
        }
    }

    /// 设置配置
    pub fn with_config(mut self, config: StreamConfig) -> Self {
        self.config = Some(config);
        self
    }

    /// 设置数据源
    pub fn with_source(mut self, source: Arc<Mutex<dyn StreamSource<Item = Message>>>) -> Self {
        self.source = Some(source);
        self
    }

    /// 设置数据汇
    pub fn with_sink(mut self, sink: Arc<Mutex<dyn StreamSink<Message>>>) -> Self {
        self.sink = Some(sink);
        self
    }

    /// 添加处理器
    pub fn add_processor(mut self, processor: Arc<dyn StreamProcessor<Message, Message>>) -> Self {
        self.processors.push(processor);
        self
    }

    /// 构建管道
    pub async fn build(self) -> StreamResult<DefaultStreamPipeline> {
        let config = self.config.unwrap_or_else(|| StreamConfig::new(&self.name));
        
        let source = self.source
            .ok_or_else(|| StreamError::config_error("未设置数据源"))?;
        
        let sink = self.sink
            .ok_or_else(|| StreamError::config_error("未设置数据汇"))?;

        let mut pipeline = DefaultStreamPipeline::new(self.name, config, source, sink);

        // 添加处理器
        for processor in self.processors {
            pipeline.add_processor(processor)?;
        }

        Ok(pipeline)
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::source::MemoryStreamSource;
    use crate::sink::MemoryStreamSink;
    use crate::processor::MapProcessor;

    #[tokio::test]
    async fn test_pipeline_creation() {
        let source = Arc::new(Mutex::new(MemoryStreamSource::new()));
        let sink = Arc::new(Mutex::new(MemoryStreamSink::new()));
        let config = StreamConfig::new("test-pipeline");

        let pipeline = DefaultStreamPipeline::new("test", config, source, sink);
        
        assert_eq!(pipeline.name(), "test");
        assert!(matches!(
            pipeline.status().await.unwrap(),
            PipelineStatus::Uninitialized
        ));
    }

    #[tokio::test]
    async fn test_pipeline_builder() {
        let source = Arc::new(Mutex::new(MemoryStreamSource::new()));
        let sink = Arc::new(Mutex::new(MemoryStreamSink::new()));

        let result = StreamPipelineBuilder::new("test-pipeline")
            .with_source(source)
            .with_sink(sink)
            .build()
            .await;

        assert!(result.is_ok());
    }
}